"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : insert into … on on duplicate key update 语句逻辑复制，有主键,full
Create At   : 2022/03/04
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.创建用户，修改pg_hba文件
    4.修改逻辑复制相关参数
    5.创建表，创建逻辑复制槽
    6.开启逻辑复制槽
    7.DML操作,insert into … on on duplicate key update 语句
    8.比对数据
    9.停止逻辑复制槽
    10.删除逻辑复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.创建成功，修改pg_hba文件成功
    4.修改逻辑复制相关参数成功
    5.创建表，创建逻辑复制槽成功
    6.开启逻辑复制槽成功
    7.DML操作执行成功
    8.比对数据成功
    9.停止逻辑复制槽成功
    10.删除逻辑复制槽成功
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.standby_node = Node('Standby1DbUser')
        self.standby_root = Node('Standby1Root')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0006/')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.user_name = 'u_standby_logic_replication_0006'
        self.table_name = 't_standby_logic_replication_0006'
        self.slot_name_1 = 'slot_standby_logic_replication_0006_1'
        self.slot_name_2 = 'slot_standby_logic_replication_0006_2'
        self.log_file_1 = os.path.join(self.dir_replication,
                                       'log_standby_logic_replication_0006_1')
        self.log_file_2 = os.path.join(self.dir_replication,
                                       'log_standby_logic_replication_0006_2')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_primy_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_primy_msg)
        self.assertEqual("OK", cp_primy_msg, "执行失败" + text)
        cp_standby_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_standby_msg)
        self.assertEqual("OK", cp_standby_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:创建用户,修改pg_hba文件;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}'; '''
        create_msg = self.sh_primy.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication all trust'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ::1/128 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step4:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:创建表，创建逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
            create table {self.table_name}( 
                col_tinyint tinyint , 
                col_smallint smallint , 
                col_integer integer, 
                col_int int, 
                col_binary_integer binary_integer, 
                col_bigint bigint, 
                col_real real, 
                col_float4 float4, 
                col_double_precision double precision, 
                col_float8 float8, 
                col_float float, 
                col_float1 float(38), 
                col_binary_double binary_double, 
                col_char char, 
                col_char1 char(50), 
                col_character character, 
                col_character1 character(50), 
                col_varchar varchar, 
                col_varchar1 varchar(50), 
                col_character_varying character varying(50), 
                col_text text, 
                col_blob blob, 
                col_bytea bytea, 
                col_date date, 
                col_time time, 
                col_time1 time(6), 
                col_time2 time without time zone,
                col_time3 time(6) without time zone, 
                col_time4 time with time zone, 
                col_time5 time(6) with time zone, 
                col_timestamp timestamp, 
                col_timestamp1 timestamp(6), 
                col_timestamp2 timestamp without time zone, 
                col_timestamp3 timestamp(6) without time zone,
                ol_timestamp4 timestamp with time zone,
                col_timestamp5 timestamp(6) with time zone);
                alter table {self.table_name} replica identity full;
                alter table {self.table_name} add constraint 
                par_p_k_0006 primary key(col_int);'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertTrue(create_msg.count(self.constant.ALTER_TABLE_MSG) == 3,
                        '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      create_msg, '执行失败:' + text)

        create_slot_1 = f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF
                        set timeout 300
                        spawn {create_slot_1}
                        expect "Password:"
                        send "{macro.COMMON_PASSWD}\\n"
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        create_slot_2 = f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                          expect <<EOF 
                          set timeout 300 
                          spawn {create_slot_2} 
                          expect "Password:" 
                          send "{macro.COMMON_PASSWD}\\n" 
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:开启逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        primy_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                          expect <<EOF 
                          set timeout 300 
                          spawn {primy_start_cmd} 
                          expect "Password:" 
                          send "{macro.COMMON_PASSWD}\\n" 
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and 't', sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        stanby_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_2} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                        expect <<EOF 
                        set timeout 300 
                        spawn {stanby_start_cmd} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_2 and 't', sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        text = "-----step7:DML操作;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"insert into {self.table_name}" \
            f"(col_integer,col_int,col_text) " \
            f"values(100,200,'only insert') " \
            f"on duplicate key update col_integer=101;"
        threads = []
        session1 = ComThread(self.sh_primy.execut_db_sql,
                             args=(sql_cmd, '', 'postgres'))
        threads.append(session1)
        session2 = ComThread(self.sh_primy.execut_db_sql,
                             args=(sql_cmd, '', 'postgres'))
        threads.append(session2)
        session3 = ComThread(self.sh_primy.execut_db_sql,
                             args=(sql_cmd, '', 'postgres'))
        threads.append(session3)
        for thread in threads:
            thread.setDaemon(True)
            thread.start()
            thread.join(20)
        for thread in threads:
            self.logger.info(thread.get_result())
            self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                          thread.get_result(), "执行失败" + text)
        time.sleep(10)

        text = "-----step8:比对数据;expect:成功-----"
        self.logger.info(text)
        shell_cmd_1 = f"sleep 10 && cat {self.log_file_1} "
        self.logger.info(shell_cmd_1)
        shell_msg_1 = self.primary_node.sh(shell_cmd_1).result()
        self.logger.info(shell_msg_1)
        self.assertNotIn("ERROR", shell_msg_1, "执行失败" + text)
        shell_cmd_2 = f"sleep 10 && cat {self.log_file_2} "
        self.logger.info(shell_cmd_2)
        shell_msg_2 = self.standby_node.sh(shell_cmd_2).result()
        self.logger.info(shell_msg_2)
        self.assertNotIn("ERROR", shell_msg_2, "执行失败" + text)
        self.assertEqual(shell_msg_1, shell_msg_2, "执行失败" + text)
        time.sleep(5)

        text = "-----step9:停止逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primy_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primy_result)
        standy_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standy_result)
        time.sleep(5)

    def tearDown(self):
        text = "-----step10:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        drop_slot_cmd_1 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-s 2 --drop"
        drop_slot_msg_1 = self.primary_node.sh(drop_slot_cmd_1).result()
        self.logger.info(drop_slot_msg_1)
        drop_slot_cmd_2 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} " \
            f"-s 2 --drop"
        drop_slot_msg_2 = self.primary_node.sh(drop_slot_cmd_2).result()
        self.logger.info(drop_slot_msg_2)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_db_cmd = f'''drop user {self.user_name};'''
        drop_db_msg = self.sh_primy.execut_db_sql(drop_db_cmd,
                                                  dbname='postgres')
        self.logger.info(drop_db_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]]; 
            then rm -rf {self.postgresql_path}; 
            mv {self.postgresql_bak_path} {self.postgresql_path}; 
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};  
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.primary_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step11.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.primary_root.sh(del_cmd)
        self.standby_root.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standby_file_msg = self.standby_root.sh(file_cmd).result()
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standby_file_msg, '执行失败' + text_3)

        self.assertEqual('', drop_slot_msg_1, '执行失败:' + text)
        self.assertEqual('', drop_slot_msg_2, '执行失败:' + text)
        self.assertNotIn(self.slot_name_1 and self.slot_name_2,
                         sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_db_msg, '执行失败' + text_1)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")